package com.zht.flink.file;

import org.apache.commons.collections.IteratorUtils;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.shaded.guava18.com.google.common.collect.Lists;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Date;

public class HotMapAnalysis {

    public static void main(String[] args) throws Exception {

        //创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        //调用FileSource
        //DataStream<String> inputStream = env.readTextFile("D:/file/123.txt");

        DataStream<String> inputStream = env.readTextFile("D:/file/Thread-01675617839967.txt");

        // 转换pojo
        DataStream<HotMapData> hotMapDataStream = inputStream.map(
                line->{
                    String[] fields = line.split("\\|");
//                    System.out.println(fields[5].substring(0, 6));
//                    System.out.println(fields[6].substring(0, 5));
//                    System.out.println(new Long(fields[1]));
                    return new HotMapData(fields[5].substring(0, 6)+ "_" + fields[6].substring(0, 5), new Long(fields[1]));
                }
        ).assignTimestampsAndWatermarks(new AscendingTimestampExtractor<HotMapData>() {
            @Override
            public long extractAscendingTimestamp(HotMapData hotMapData) {
                return hotMapData.getDataTime();
            }
        });

        //分组开窗聚合
//        DataStream<HotMapResult> hotMapResultAggStream =
//                hotMapDataStream.keyBy("address").countWindow(100).aggregate(new HotMapCountAgg(), new WindowHotMapCountResult());
        DataStream<HotMapResult> hotMapResultAggStream =
                hotMapDataStream.keyBy("address").timeWindow(Time.seconds(500), Time.milliseconds(5000)).aggregate(new HotMapCountTimeAgg(), new WindowHotMapCountTimeResult());

        //收集同一窗框的所有(经纬度：count)数据, 输出
        DataStream <String> stringDataStream = hotMapResultAggStream.keyBy("windowEnd")
                .process(new HotItemOneWindow());

        stringDataStream.print();
        env.execute("HotMap");
    }

    public static class HotMapCountTimeAgg implements AggregateFunction<HotMapData, Long, Long>{
        @Override
        public Long createAccumulator() {
            return 0L;
        }

        @Override
        public Long add(HotMapData hotMapData, Long accumulator) {
            return accumulator + 1;
        }

        @Override
        public Long getResult(Long accumulator) {
            return accumulator;
        }

        @Override
        public Long merge(Long a, Long b) {
            return a + b;
        }
    }

    //自定义全窗口函数
    /*
        1.增量聚合的结果
     */
    public static class WindowHotMapCountTimeResult implements WindowFunction<Long, HotMapResult, Tuple, TimeWindow> {
        @Override
        public void apply(Tuple tuple, TimeWindow timeWindow, Iterable<Long> iterable, Collector<HotMapResult> collector) throws Exception {
            String address = tuple.getField(0);
            // 窗口的结束时间
            Long windowEnd = timeWindow.getEnd();
            Long count = iterable.iterator().next();
            // Long count = new  Long (IteratorUtils.toList(iterable.iterator()).size());
            collector.collect(new HotMapResult(address, windowEnd, count));
        }
    }

    /*
        1. keyby
        2. 上一步的聚合结果
        3. 输出格式
     */
    public static class HotItemOneWindow extends KeyedProcessFunction<Tuple, HotMapResult, String> {

        // 定义状态
        ListState<HotMapResult> hotMapResultListState;

        @Override
        public void open(Configuration parameters) throws Exception {
            hotMapResultListState = getRuntimeContext().getListState(new ListStateDescriptor<HotMapResult>("hotMapResultListState",HotMapResult.class ));
        }

        @Override
        public void processElement(HotMapResult hotMapResult, Context context, Collector<String> collector) throws Exception {
            hotMapResultListState.add(hotMapResult);
            context.timerService().registerEventTimeTimer(hotMapResult.getWindowEnd() + 1);
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            //排序输出
            ArrayList<HotMapResult> hotMapResults= Lists.newArrayList(hotMapResultListState.get().iterator());
            // hotMapResults.sort();
            StringBuilder stringBuilder = new StringBuilder();
            stringBuilder.append("========================================");
            stringBuilder.append("窗口结束时间：").append(new Timestamp(timestamp - 1 )).append("\n");

            //遍历
            for (int i = 0; i < hotMapResults.size(); i++) {
                HotMapResult hotMapResult = hotMapResults.get(i);
                stringBuilder.append(hotMapResult);
                stringBuilder.append("\n");
            }
            stringBuilder.append("========================================");

            Thread.sleep(1000l);
            out.collect(stringBuilder.toString());
        }
    }



    public static class Splitter implements FlatMapFunction<String, Tuple2<String, Integer>> {
        @Override
        public void flatMap(String sentence, Collector<Tuple2<String, Integer>> out) throws Exception {
            for (String word: sentence.split(" ")) {
                out.collect(new Tuple2<String, Integer>(word, 1));
            }
        }
    }

}